/*
 * @Author: ygnnli 1585923072@qq.com
 * @Date: 2024-08-20 17:24:01
 * @LastEditors: ygnnli 1585923072@qq.com
 * @LastEditTime: 2024-08-27 22:53:18
 * @FilePath: /bitrpc/source/client/rpc_topic.hpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#pragma once
#include "requestor.hpp"
#include <unordered_set>

namespace bitrpc
{
    namespace client
    {
        class TopicManger
        {
            public:
            using SubCallback = std::function<void(const std::string &key,const std::string &msg)>; //订阅回调函数
            using ptr = std::shared_ptr<TopicManger>;
            TopicManger(const Requestor::ptr &requestor)
                :_requestor(requestor)
            {}
            //创建主题
            bool create(const BaseConnection::ptr &conn,const std::string &key)
            {
                return commonRequest(conn,key,TopicOpType::TOPIC_CREATE);
            }

            //删除主题
            bool remove(const BaseConnection::ptr &conn,const std::string &key)
            {
                return commonRequest(conn,key,TopicOpType::TOPIC_REMOVE);
            }
            //订阅主题
            bool subscribe(const BaseConnection::ptr &conn,const std::string &key,const SubCallback &cb)
            {
                //放在前面的原因:如果刚刚订阅成功就有消息发过来,此时还没有创建好回调函数,所以需要先创建好回调函数
                addSubscribe(key,cb);
                bool ret = commonRequest(conn,key,TopicOpType::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    delSubscribe(key);
                    return false;
                }
                return true;
            }
            //取消订阅主题
            bool cancel(const BaseConnection::ptr &conn,const std::string &key)
            {
                delSubscribe(key);
                return commonRequest(conn,key,TopicOpType::TOPIC_CANCEL);
            }
            //发布消息
            bool publish(const BaseConnection::ptr &conn,const std::string &key,const std::string &msg)
            {
                return commonRequest(conn,key,TopicOpType::TOPIC_PUBLISH,msg);
            }

            //订阅后:主题操作请求!!
            void onPublish(const BaseConnection::ptr& conn,const TopicRequest::ptr& msg)
            {
                //1. 从消息中取出操作类型进行判断，是否是消息请求
                auto type = msg->optype();
                if(type != TopicOpType::TOPIC_PUBLISH)
                {
                    ELOG("收到错误类型的主题操作!");
                    return;
                }
                //2. 取出消息主题名称，以及消息内容
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                //3. 通过主题名称，查找对应主题的回调处理函数，有在处理，无在报错
                auto callback = getSubscribe(topic_key);//获取主题名称对应的回调函数
                if(!callback)
                {
                    ELOG("收到了 %s 主题消息,但是该消息没有主题处理回调!",topic_key.c_str());
                    return;
                }
                //4. 调用回调函数，处理消息
                return callback(topic_key,topic_msg);
            }

            private:
            //添加主题名称和回调函数的映射
            void addSubscribe(const std::string &key,const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key,cb));//将回调函数插入到回调函数集合中
            }
            //删除主题名称和回调函数的映射
            void delSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }

            //获取主题名称对应的回调函数
            const SubCallback getSubscribe(const std::string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if(it == _topic_callbacks.end())
                {
                    ELOG("主题名称不存在!");
                    return SubCallback();//返回一个空的回调函数:临时对象具有常性需要const
                }
                return it->second;
            }
            //请求主题操作
            bool commonRequest(const BaseConnection::ptr &conn,const std::string &key,TopicOpType type,const std::string &msg = "")
            {
                //1. 构造请求对象，并填充数据
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setMType(MType::REQ_TOPIC);
                msg_req->setOptype(type);//创建主题
                msg_req->setTopicKey(key);//主题名称
                //1.1 如果是发布消息，则填充消息内容
                if(type == TopicOpType::TOPIC_PUBLISH)
                {
                    ELOG("发布消息主题!!!");
                    msg_req->setTopicMsg(msg);
                }
                //2. 向服务端发送请求，等待响应
                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn, msg_req, msg_rsp);
                if(ret == false)
                {
                    ELOG("主题操作请求失败!");
                    return false;
                }
                //3. 判断请求处理是否成功   
                auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if(!topic_rsp_msg)
                {
                    ELOG("主题操作响应,向下类型转换失败!");
                    return false;
                }
                if(topic_rsp_msg->rcode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错: %s",errReason(topic_rsp_msg->rcode()).c_str());
                    return false;
                }
                return true;
            }

            private:
            std::mutex _mutex;
            std::unordered_map<std::string,SubCallback> _topic_callbacks;//主题名称和回调函数的映射
            Requestor::ptr _requestor;
        };
    }
}